package queue;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Bounded blocking queue
 */
public class BoundedQueueSplit<T> implements Queue<T> {
    private static int counter = 0;
    // Lock out other enqueuers (dequeuers)
    ReentrantLock enqLock, deqLock;
    /**
     * wait/signal when queue is not empty or not full
     */
    Condition notEmptyCondition, notFullCondition;
    /**
     * Number of empty slots.
     */
    AtomicInteger size;
    AtomicInteger deqSize;
    AtomicInteger enqSize;
    /**
     * First entry in queue.
     */
    Entry head;
    /**
     * Last entry in queue.
     */
    Entry tail;
    /**
     * Max number of entries allowed in queue.
     */
    int capacity;

    public int getCapacity() {
        return capacity;
    }
    /**
     * Constructor.
     * 
     * @param capacity
     *            Max number of items allowed in queue.
     */
    public BoundedQueueSplit(int capacity) {
        this.capacity = capacity;
        this.head = new Entry(null);
        this.tail = head;
        this.size = new AtomicInteger(0);
        this.deqSize = new AtomicInteger(0);
        this.enqSize = new AtomicInteger(0);
        this.enqLock = new ReentrantLock();
        this.notFullCondition = enqLock.newCondition();
        this.deqLock = new ReentrantLock();
        this.notEmptyCondition = deqLock.newCondition();
        System.out.println("time: [" + counter++ +"] Thread: [" + Thread.currentThread().getId() + "]: Created bounded queue with capacity [" + capacity+ "]");
        
    }

    /**
     * Remove and return head of queue.
     * 
     * @return remove first item in queue
     */
    public T deq() {
        T result;
        boolean mustWakeEnqueuers = false;
        System.out.println("time: [" + counter++ +"] Thread: [" + Thread.currentThread().getId() 
                + "]: Acquired Lock for dequeing an element.");
        
        deqLock.lock();
        try {
            while ((enqSize.get() - deqSize.get()) == 0) { // 如果队列为空的状态
                try {
                    System.out.println("time: [" + counter++ +"] Thread: [" + Thread.currentThread().getId()
                            + "] waiting for notEmptyCondition.");
                    notEmptyCondition.await(); // 等待其他线程的入队操作
                } catch (InterruptedException ex) {
                }
            }
            result = head.next.value;
            head = head.next;
            if ((enqSize.get() - deqSize.get()) == capacity) { // 队列为满的状态

                mustWakeEnqueuers = true; // 唤醒入队者的标记
            }
            deqSize.getAndIncrement();
        } finally {
            deqLock.unlock();
        }
        if (mustWakeEnqueuers) { // 唤醒入队者
            enqLock.lock();
            try {

                System.out.println("time: [" + counter++ +"] Thread: [" + Thread.currentThread().getId()
                        + "] Signalling not full condition after a dequeue operation.");
                notFullCondition.signalAll(); // enqLock 发出队列没满的信号
            } finally {
                enqLock.unlock();
            }
        }
        return result;
    }

    /**
     * Append item to end of queue.
     * 
     * @param x
     *            item to append
     */
    public void enq(T x) {
        if (x == null)
            throw new NullPointerException();
        boolean mustWakeDequeuers = false;
        System.out.println("time: [" + counter++ +"] Thread: [" + Thread.currentThread().getId() 
                + "]: Acquired Lock for enqueing an element.");
        
        enqLock.lock();
        try {
            while ((enqSize.get() - deqSize.get()) == capacity) { // 发现队列满了
                try {
                    System.out.println("time: [" + counter++ +"] Thread: [" + Thread.currentThread().getId()
                            + "] waiting for notFullCondition.");
                    notFullCondition.await(); // 等待没满的信号，这里会释放锁
                } catch (InterruptedException e) {
                }
            }
            Entry e = new Entry(x);
            tail.next = e;
            tail = e;
            if ((enqSize.get() - deqSize.get()) == 0) { // 修改之后size = 1，此时队列不在为空可以deque
                mustWakeDequeuers = true;
            }
            enqSize.getAndIncrement();
        } finally {
            enqLock.unlock();
        }
        if (mustWakeDequeuers) {
            deqLock.lock();
            try {

                System.out.println("Ttme: [" + counter++ +"] Thread: [" + Thread.currentThread().getId()
                        + "] Signalling not empty condition after a enqueue operation.");
                notEmptyCondition.signalAll(); // deqLock 不空
            } finally {
                deqLock.unlock();
            }
        }
    }

    /**
     * Individual queue item.
     */
    protected class Entry {
        /**
         * Actual value of queue item.
         */
        public T value;
        /**
         * next item in queue
         */
        public Entry next;

        /**
         * Constructor
         * 
         * @param x
         *            Value of item.
         */
        public Entry(T x) {
            value = x;
            next = null;
        }
    }
}
